package com.lagou.rabbitmq.publisherconfirm;

import com.rabbitmq.client.*;

import java.nio.charset.StandardCharsets;

public class PublisherConfirmProducer2 {

    public static void main(String[] args) throws Exception {

        ConnectionFactory factory = new ConnectionFactory();
        //指定协议 amqp://
        //指定用户名 root
        //指定密码 123456
        //指定host 47.115.74.219
        //指定端口号 5672
        //指定虚拟主机 %2f
        factory.setUri("amqp://root:123456@47.115.74.219:5672/%2f");

        Connection connection = factory.newConnection();

        Channel channel = connection.createChannel();

        //向RabbitMQ服务器发送AMQP命令，将当前通道标记为发送方确认通道
        AMQP.Confirm.SelectOk selectOk = channel.confirmSelect();

        channel.queueDeclare("queue.pc",true,false,false,null);
        channel.exchangeDeclare("ex.pc", BuiltinExchangeType.DIRECT,true,false,null);
        channel.queueBind("queue.pc","ex.pc","key.pc");

        String message = "hello-";
        //批处理大小
        int batchSize = 10;
        //用于对需要等待确认消息的计数
        int outstrandingConfirms = 0;
        for(int i = 0; i < 103;i++) {
            channel.basicPublish("ex.pc", "key.pc", null, (message+i).getBytes(StandardCharsets.UTF_8));
            outstrandingConfirms++;
            if(outstrandingConfirms == batchSize){
                //此时已经有一个批次的消息需要同步等待broker的确认消息
                //同步等待
                channel.waitForConfirmsOrDie(5_000);
                System.out.println("消息已经确认");
                outstrandingConfirms = 0;
            }
        }
        if(outstrandingConfirms > 0){
            channel.waitForConfirmsOrDie(5_000);
            System.out.println("剩余消息已经确认");
        }

        channel.close();
        connection.close();

    }

}
